package com.hadwinling.alogriithm.streaming

import org.apache.spark.SparkConf
import org.apache.spark.streaming.dstream.{DStream, ReceiverInputDStream}
import org.apache.spark.streaming.{Seconds, StreamingContext}

object SparkStreaming01_WordCount  {
  def main(args: Array[String]): Unit = {
    // TODO 创建环境对象
    // StreamingContext创建时，需要传递两个参数
    // 第一个参数表示环境配置
    val conf= new SparkConf().setMaster("local[*]").setAppName("Streaming")
    // 第二个参数表示批量处理的周期（采集周期）
    val ssc = new StreamingContext(conf,Seconds(3))



    // todo 逻辑处理
    // 获取端口数据
    //3.通过监控端口创建 DStream,读进来的数据为一行行
    val lines: ReceiverInputDStream[String] = ssc.socketTextStream("localhost", 9999)

    //将每一行数据做切分,形成一个个单词
    val words = lines.flatMap(_.split(" "))

    //将单词映射成元组(word,1)
    val wordTOone = words.map((_,1))

    //将相同的单词次数做统计
    val wordToCount :DStream[(String,Int)] = wordTOone.reduceByKey(_+_)
    //打印
    wordToCount.print()


    // 由于SparkStreaming采集器是长期执行的任务，所以不能直接关闭
    // 如果main方法执行完毕，应用程序也会自动结束。所以不能让main执行完毕
    //ssc.stop()
    // 1. 启动采集器
    ssc.start()
    // 2. 等待采集器的关闭
    ssc.awaitTermination()
/*
    启动程序并通过 netcat 发送数据:
    nc -lk 9999
    hello spark
*/

  }
}
